package com.gzyj.flink.gps.map.roadlevel;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.gzyj.flink.dto.alarm.Kafka_VehicleRoadLevelAlarm;
import com.gzyj.flink.dto.alarm.MonitorClientAlarm;
import com.gzyj.flink.gps.config.RedisKeyPrefixConfig;
import com.gzyj.flink.gps.factory.MySQLMybatisSessionFactory;
import com.gzyj.flink.gps.factory.RedisSessionFactory;
import com.gzyj.flink.gps.mapper.mysql.MonYanzhongweiguiSetMapper;
import com.gzyj.flink.gps.service.YjGpsAddrService;
import com.gzyj.flink.gps.utils.TimeUtils;
import com.gzyj.flink.pojo.gps.BaobiaoAlarmsummary;
import com.gzyj.flink.pojo.gps.BaobiaoAlarmsummaryRecord;
import com.gzyj.flink.pojo.mysql.set.MonYanzhongweiguiSet;
import com.gzyj.flink.veh.VehicleIncrement;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.Set;

public class KafkaRoadLevelAlarmProcess extends RichMapFunction<Kafka_VehicleRoadLevelAlarm, KafkaProcessResult> {
    LettuceConnectionFactory redisFactory;
    YjGpsAddrService yjGpsAddrService;
    SqlSessionFactory mysqlSessionFactory;

    private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisSessionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

    public VehicleIncrement getVehicleInfo(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {

        Set<String> keys = clusterRedisTemplate.keys(RedisKeyPrefixConfig.RedisKey_VehicleInfo + "*:" + vehId + "*");
        VehicleIncrement vehicleInfo = null;
        if (CollectionUtil.isNotEmpty(keys)) {
            Object obj = clusterRedisTemplate.opsForValue().get(new ArrayList<String>(keys).get(0));
            if (obj == null) {
                return null;
            } else {
                vehicleInfo = JSONObject.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), VehicleIncrement.class);
            }
        } else {
            return null;
        }
        return vehicleInfo;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        redisFactory = RedisSessionFactory.getRedisFactory();
        yjGpsAddrService = new YjGpsAddrService();
        mysqlSessionFactory = MySQLMybatisSessionFactory.getSqlSessionFactory();
    }

    public MonYanzhongweiguiSet getAlarmDeploy(String vehId, MonYanzhongweiguiSetMapper monYanzhongweiguiSetMapper) {
        return monYanzhongweiguiSetMapper.selectByVehId(vehId);
    }

    @Override
    public KafkaProcessResult map(Kafka_VehicleRoadLevelAlarm RoadLevelAlarm) throws Exception {
        RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(redisFactory);

        SqlSession mysql = mysqlSessionFactory.openSession(true);
        KafkaProcessResult tuple3 = new KafkaProcessResult();
        MonYanzhongweiguiSetMapper monYanzhongweiguiSetMapper = mysql.getMapper(MonYanzhongweiguiSetMapper.class);
        try {

            VehicleIncrement vehinfo = getVehicleInfo(RoadLevelAlarm.getVehId(), clusterRedisTemplate);

            if (vehinfo == null) return null;
            if (StringUtils.isEmpty(RoadLevelAlarm.getRoadName())) {
                String road = yjGpsAddrService.getAddress(RoadLevelAlarm.getLatitude(), RoadLevelAlarm.getLongitude());
                if (StringUtils.isNotEmpty(road)) {
                    RoadLevelAlarm.setRoadName(road);
                }
            }


            if (RoadLevelAlarm.getAlarmFlag() == 1) {
                // 推送到kafka
                MonitorClientAlarm vi = new MonitorClientAlarm();
                vi.VehNo = vehinfo.getCheliangpaizhao();
                vi.VehId = RoadLevelAlarm.getVehId();
                vi.PlateColor = vehinfo.getChepaiyanse();
                vi.AlarmFlag = 1;
                vi.Time = RoadLevelAlarm.getTime();
                vi.Note = "超速报警";
                vi.Longitude = RoadLevelAlarm.getLongitude();
                vi.Latitude = RoadLevelAlarm.getLatitude();
                vi.Velocity = RoadLevelAlarm.getVelocity();
                vi.Angle = RoadLevelAlarm.getAngle();
                vi.Local = RoadLevelAlarm.getLocal();
                vi.Limited = RoadLevelAlarm.getRoadLimited();
                vi.RoadLimited = RoadLevelAlarm.getRoadLimited();
                vi.RoadName = RoadLevelAlarm.getRoadName();
                vi.AlarmGuid = RoadLevelAlarm.getAlarmGuid();
                vi.Driver = RoadLevelAlarm.getDriver();
                vi.alarmSource = 1;
                vi.AlarmLevel = RoadLevelAlarm.getAlarmLevel();

                tuple3.setMonitorClientAlarm(vi);

                //记录开始报警点
                clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_CmAlarm + RoadLevelAlarm.getVehId() + "超速", JSON.toJSONString(RoadLevelAlarm, SerializerFeature.DisableCircularReferenceDetect));
                BaobiaoAlarmsummary baobiaoAlarmsummary = new BaobiaoAlarmsummary();
                baobiaoAlarmsummary.setVehid(RoadLevelAlarm.getVehId());
                baobiaoAlarmsummary.setDeptId(RoadLevelAlarm.getDeptId());
                baobiaoAlarmsummary.setBegintime(TimeUtils.toDate(RoadLevelAlarm.getTime()));
                baobiaoAlarmsummary.setDriverName(RoadLevelAlarm.getDriver());
                baobiaoAlarmsummary.setDriverNo(RoadLevelAlarm.getDriverNo());
                baobiaoAlarmsummary.setMaxspeed(Double.valueOf(RoadLevelAlarm.getVelocity()).intValue());
                baobiaoAlarmsummary.setAlarmtype("超速报警");
                baobiaoAlarmsummary.setLongitude(BigDecimal.valueOf(RoadLevelAlarm.getLongitude()));
                baobiaoAlarmsummary.setLatitude(BigDecimal.valueOf(RoadLevelAlarm.getLatitude()));
                baobiaoAlarmsummary.setVelocity(Double.valueOf(RoadLevelAlarm.getVelocity()).intValue());
                baobiaoAlarmsummary.setAngle(RoadLevelAlarm.getAngle());
                baobiaoAlarmsummary.setLocal(RoadLevelAlarm.getLocal());
                baobiaoAlarmsummary.setElevation(BigDecimal.ZERO);
                baobiaoAlarmsummary.setLimited(RoadLevelAlarm.getRoadLimited());
                baobiaoAlarmsummary.setTime(TimeUtils.toDate(RoadLevelAlarm.getTime()));
                baobiaoAlarmsummary.setRoadName(RoadLevelAlarm.getRoadName());
                baobiaoAlarmsummary.setRoadLevel("");
                baobiaoAlarmsummary.setRoadLimited(RoadLevelAlarm.getRoadLimited());
                baobiaoAlarmsummary.setDisposealarmname("");
                baobiaoAlarmsummary.setDisposealarmtime(TimeUtils.toDate(RoadLevelAlarm.getTime()));
                baobiaoAlarmsummary.setAlarmcl("短信发送");
                baobiaoAlarmsummary.setAnalyzemode(1);
                baobiaoAlarmsummary.setStatus(0);
                baobiaoAlarmsummary.setSyntime(new Date());
                baobiaoAlarmsummary.setPlatenumber(vehinfo.getCheliangpaizhao());
                baobiaoAlarmsummary.setColor(vehinfo.getChepaiyanse());
                baobiaoAlarmsummary.setCompany(vehinfo.getDeptName());
                baobiaoAlarmsummary.setOperattype(vehinfo.getShiyongxingzhi());
                baobiaoAlarmsummary.setAlarmguid(RoadLevelAlarm.getAlarmGuid());

                tuple3.setBaobiaoAlarmsummary(baobiaoAlarmsummary);

                BaobiaoAlarmsummaryRecord baobiaoAlarmsummaryRecord = new BaobiaoAlarmsummaryRecord();
                baobiaoAlarmsummaryRecord.setVehid(RoadLevelAlarm.getVehId());
                baobiaoAlarmsummaryRecord.setBegintime(TimeUtils.toDate(RoadLevelAlarm.getTime()));
                baobiaoAlarmsummaryRecord.setAlarmtype("超速报警");
                baobiaoAlarmsummaryRecord.setLongitude(BigDecimal.valueOf(RoadLevelAlarm.getLongitude()));
                baobiaoAlarmsummaryRecord.setLatitude(BigDecimal.valueOf(RoadLevelAlarm.getLatitude()));
                baobiaoAlarmsummaryRecord.setVelocity(Double.valueOf(RoadLevelAlarm.getVelocity()).intValue());
                baobiaoAlarmsummaryRecord.setAngle(RoadLevelAlarm.getAngle());
                baobiaoAlarmsummaryRecord.setRoadLevel(RoadLevelAlarm.getRoadLevel());
                baobiaoAlarmsummaryRecord.setRoadLimited(RoadLevelAlarm.getRoadLimited());
                baobiaoAlarmsummaryRecord.setKeeptime(0);
                baobiaoAlarmsummaryRecord.setLevel(1);
                baobiaoAlarmsummaryRecord.setAlarmreportid(RoadLevelAlarm.getAlarmGuid());

                tuple3.setBaobiaoAlarmsummaryRecord(baobiaoAlarmsummaryRecord);

            } else {
                // 获取开始报警点
                Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_CmAlarm + RoadLevelAlarm.getVehId() + "超速");
                if (obj == null) return null;
                String jsonvalue = (String) obj;
                Kafka_VehicleRoadLevelAlarm LastRoadLeave = JSON.parseObject(jsonvalue, Kafka_VehicleRoadLevelAlarm.class);
                // 计算超速比:ChaoSuBi,持续时间:KeepTime
                int ChaoSuBi = (int) Math.round((RoadLevelAlarm.getMaxSpeed() - LastRoadLeave.getRoadLimited()) * 100.0 / LastRoadLeave.getRoadLimited());
                if (LastRoadLeave.getRoadLimited() == 0) ChaoSuBi = 0;

                double KeepTime = TimeUtils.getMinusSecond(RoadLevelAlarm.getAlarmEndTime(), RoadLevelAlarm.getAlarmBeginTime());
                if (KeepTime < 0) KeepTime = 0;
                // 计算严重违规等级  报警等级:status 默认为1级
                int cssj = -1;
                int cssj1 = -1;
                int cssj2 = -1;
                double csbfb = -1;
                double csbfb1 = -1;
                double csbfb2 = -1;
                MonYanzhongweiguiSet AlarmDeploy = getAlarmDeploy(RoadLevelAlarm.getVehId(), monYanzhongweiguiSetMapper);
                if (AlarmDeploy != null) {
                    cssj = AlarmDeploy.getCscxsj() == null ? -1 : AlarmDeploy.getCscxsj();
                    cssj1 = AlarmDeploy.getCscxsj1() == null ? -1 : AlarmDeploy.getCscxsj1();
                    cssj2 = AlarmDeploy.getCscxsj2() == null ? -1 : AlarmDeploy.getCscxsj2();
                    csbfb = AlarmDeploy.getCsbfb() == null ? -1 : AlarmDeploy.getCsbfb();
                    csbfb1 = AlarmDeploy.getCsbfb1() == null ? -1 : AlarmDeploy.getCsbfb1();
                    csbfb2 = AlarmDeploy.getCsbfb2() == null ? -1 : AlarmDeploy.getCsbfb2();

                }

                int status = 1;
                if (cssj1 > 0 && csbfb1 > 0) {
                    if (KeepTime >= cssj1 || ChaoSuBi >= csbfb1) status = 2;
                }
                if (cssj2 > 0 && csbfb2 > 0) {
                    if (KeepTime >= cssj2 || ChaoSuBi >= csbfb2) status = 3;
                }

                if (status < RoadLevelAlarm.getAlarmLevel()) {
                    status = RoadLevelAlarm.getAlarmLevel();
                }

                BaobiaoAlarmsummary baobiaoAlarmsummary = new BaobiaoAlarmsummary();
                baobiaoAlarmsummary.setEndtime(TimeUtils.date2Str(RoadLevelAlarm.getAlarmEndTime()));
                baobiaoAlarmsummary.setMaxspeed(Double.valueOf(RoadLevelAlarm.getMaxSpeed()).intValue());

                if (AlarmDeploy != null) {
                    if (KeepTime >= 1 && LastRoadLeave.getVelocity() > 30) {
                        baobiaoAlarmsummary.setPassed(100);
                    } else {
                        baobiaoAlarmsummary.setPassed(-1);
                    }
                } else {
                    if (KeepTime >= 1 && LastRoadLeave.getVelocity() > 30) {
                        status = 2;
                        baobiaoAlarmsummary.setPassed(100);
                    } else {
                        baobiaoAlarmsummary.setPassed(-1);
                    }
                }

                baobiaoAlarmsummary.setKeeptime(KeepTime);
                baobiaoAlarmsummary.setVerifyname("System");
                baobiaoAlarmsummary.setVerifytime(new Date());
                baobiaoAlarmsummary.setEndspeed(Double.valueOf(RoadLevelAlarm.getVelocity()).intValue());
                baobiaoAlarmsummary.setDistance(Double.valueOf(RoadLevelAlarm.getDistance()));
                baobiaoAlarmsummary.setCutofftime(new Date());
                baobiaoAlarmsummary.setEndlongitude(BigDecimal.valueOf(RoadLevelAlarm.getLongitude()));
                baobiaoAlarmsummary.setEndlatitude(BigDecimal.valueOf(RoadLevelAlarm.getLatitude()));
                baobiaoAlarmsummary.setEndroadname(RoadLevelAlarm.getRoadName());
                baobiaoAlarmsummary.setIsregionv("0");
                baobiaoAlarmsummary.setAtnocturnal(0);
                baobiaoAlarmsummary.setBaojingtype(1);
                baobiaoAlarmsummary.setChaosubi(ChaoSuBi);
                baobiaoAlarmsummary.setIcardsign(RoadLevelAlarm.getDriver());
                baobiaoAlarmsummary.setStatus(status);

                baobiaoAlarmsummary.setVehid(RoadLevelAlarm.getVehId());
                baobiaoAlarmsummary.setDeptId(RoadLevelAlarm.getDeptId());
                baobiaoAlarmsummary.setAnalyzemode(1);
                baobiaoAlarmsummary.setAlarmtype("超速报警");
                baobiaoAlarmsummary.setBegintime(RoadLevelAlarm.getAlarmBeginTime());

                tuple3.setBaobiaoAlarmsummary(baobiaoAlarmsummary);


                BaobiaoAlarmsummaryRecord baobiaoAlarmsummaryRecord = new BaobiaoAlarmsummaryRecord();
                baobiaoAlarmsummaryRecord.setVehid(RoadLevelAlarm.getVehId());
                baobiaoAlarmsummaryRecord.setBegintime(TimeUtils.toDate(RoadLevelAlarm.getTime()));
                baobiaoAlarmsummaryRecord.setAlarmtype("超速报警");
                baobiaoAlarmsummaryRecord.setLongitude(BigDecimal.valueOf(RoadLevelAlarm.getLongitude()));
                baobiaoAlarmsummaryRecord.setLatitude(BigDecimal.valueOf(RoadLevelAlarm.getLatitude()));
                baobiaoAlarmsummaryRecord.setVelocity(Double.valueOf(RoadLevelAlarm.getVelocity()).intValue());
                baobiaoAlarmsummaryRecord.setAngle(RoadLevelAlarm.getAngle());
                baobiaoAlarmsummaryRecord.setRoadLevel(RoadLevelAlarm.getRoadLevel());
                baobiaoAlarmsummaryRecord.setRoadLimited(LastRoadLeave.getRoadLimited());
                baobiaoAlarmsummaryRecord.setKeeptime(Double.valueOf(KeepTime).intValue());
                baobiaoAlarmsummaryRecord.setLevel(status);
                baobiaoAlarmsummaryRecord.setAlarmreportid(RoadLevelAlarm.getAlarmGuid());

                tuple3.setBaobiaoAlarmsummaryRecord(baobiaoAlarmsummaryRecord);


                clusterRedisTemplate.delete(RedisKeyPrefixConfig.RedisKey_CmAlarm + RoadLevelAlarm.getVehId() + "超速");
            }

        } catch (Exception e) {
            e.printStackTrace();
            mysql.rollback();
        } finally {
            mysql.close();
        }

        return tuple3;
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
